package com.sali;

import com.sali.commons.config.TaskConfig;
import com.sali.commons.config.ThreadNamedFactory;
import com.sali.commons.config.ThreadPoolManager;
import com.sali.entity.TransferObject;
import com.sali.service.data_entry.FileReadListener;
import com.sali.service.data_file_uploader.refcount.ReferenceCountMonitor;
import com.sali.service.data_processor.DataProcessor;
import com.sali.service.data_processor.ProcessThread;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class TaskApplication {

    private static final Logger logger = LoggerFactory.getLogger(TaskApplication.class);

    // data entry到data processor的数据管道
    private static BlockingQueue<List<TransferObject>> dataChannel = new ArrayBlockingQueue<>(5000000);


    public static void main(String[] args) {
        Thread.currentThread().setName("task-main->" + TaskApplication.class.getSimpleName());

        // 初始化数据入口
        FileReadListener fileReadListener = new FileReadListener(TaskConfig.DATA_ENTRY_FILE_PATH, dataChannel);
        // 初始化定时调度线程池
        ThreadPoolManager.initScheduledExecutor();
        // 初始化引用计数监听线程
        ThreadPoolManager.getScheduledExecutor().scheduleAtFixedRate(new ReferenceCountMonitor(), 0, TaskConfig.REFERENCE_COUNT_INTERVAL, TimeUnit.MILLISECONDS);

        // 执行数据入口监听
        fileReadListener.start();
        // 执行数据处理过程
        ThreadPoolManager.setRealtimeThreadPool(TaskConfig.REALTIME_PROCESS_THREAD_NUM, TaskConfig.REALTIME_PROCESS_THREAD_NUM, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new ThreadNamedFactory("realtime-task-process-pool-"));
        final String realtimeConfigFilePath = TaskConfig.REALTIME_PROCESS_CONFIG_PATH;
        DataProcessor realTimeDataProcessor = new DataProcessor(realtimeConfigFilePath);
        for (int i = 0; i < TaskConfig.REALTIME_PROCESS_THREAD_NUM; i++) {
            final ProcessThread processThread = new ProcessThread(realTimeDataProcessor, dataChannel);
            logger.info("[任务主线程] 启动Task处理线程" + i);
            processThread.setName("realtime-task-process-" + ProcessThread.class.getSimpleName() + "-" + i);
            ThreadPoolManager.executeRealtimeThreadPool(processThread);
        }

        //JVM 关闭时调用
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            logger.info("[任务主线程] 主线程停止中.........");
            ThreadPoolManager.shutdownScheduledExecutor();
        }));
    }
}